package com.athui.utils.explorer.metadata.impl;

import com.athui.bean.metadata.ColumnMetaData;
import com.athui.bean.config.DataType;
import com.athui.bean.metadata.TableMetaData;
import com.athui.utils.common.ConfigureUtils;
import com.athui.utils.common.RegexMatchUtils;
import com.athui.utils.common.jdbc.JdbcResultSetBuilder;
import com.athui.utils.explorer.metadata.MetaStoreExplorer;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.tuple.Tuple3;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * @description: TODO Hive 元数据获取 探查
 * @autor: zhangzhonghui
 * @create: 2024-08-16 20:59
 * @Version: 1.0
 */
public class HiveMetaStoreExplorer  extends MetaStoreExplorer {
    @Override
    public Map<String, DataType> typeMapping() {

        Map<String, DataType> typeMapping = new HashMap<>();
        // 字符型
        typeMapping.put("string", new DataType(BasicTypeInfo.STRING_TYPE_INFO,DataType.TEXT));
        // 数值型
        typeMapping.put("decimal", new DataType(BasicTypeInfo.BIG_DEC_TYPE_INFO, DataType.NUMERICAL));


        return typeMapping;
    }

    @Override
    public TableMetaData tableMetaDataBuilder(String schemaName, String tableName) {

        return super.operation(new JdbcResultSetBuilder<TableMetaData>() {
            @Override
            public String buildSql() {
                Map<String, Object> params =new HashMap<>();
                params.put("schema_name",schemaName);
                params.put("table_name",tableName);

                return ConfigureUtils.mapperParser("hiveTableMetaStore", params).toString();
            }

            @Override
            public TableMetaData accept(ResultSet resultSet) throws SQLException {
                TableMetaData tableMetaData = new TableMetaData();
                while ( resultSet.next()){
                    // 库名
                    tableMetaData.setSchemaName(resultSet.getString("schema_name"));
                    // 表名
                    tableMetaData.setTableName(resultSet.getString("table_name"));
                    // 表类型
                    tableMetaData.setTableType(resultSet.getString("table_type"));
                    // 表注释
                    tableMetaData.setTableComment(resultSet.getString("table_comment"));
                    // 表的创建时间
                    tableMetaData.setCreateTime(resultSet.getString("create_time"));
                    // 表的最后更新时间
                    tableMetaData.setLastUpdateTime(resultSet.getString("last_update_time"));
                    // 是否为外部表
                    tableMetaData.setExternal("TRUE".equals(resultSet.getString("is_external")));
                    // 输入格式
                    tableMetaData.setInputFormat(resultSet.getString("input_format"));
                    // 输出格式
                    tableMetaData.setOutputFormat(resultSet.getString("output_format"));
                    // 表的存储路径
                    tableMetaData.setLocation(resultSet.getString("location"));
                    // 是否为分区表
                    tableMetaData.setPartitioned("TRUE".equals(resultSet.getString("is_partitioned")));
                    // 分区数
                    tableMetaData.setPartitionedSize(resultSet.getInt("partitioned_size"));
                    // 分区列
                    tableMetaData.setPartitionedColumns(resultSet.getString("partitioned_columns"));
                    // 分桶数
                    tableMetaData.setBucketsSize(resultSet.getInt("buckets_size"));
                }

                return tableMetaData;
            }
        });
    }

    @Override
    public String queryColumnMetaDataSql(String schemaName, String tableName) {
        Map<String, Object> params =new HashMap<>();
        params.put("schema_name",schemaName);
        params.put("table_name",tableName);

        return ConfigureUtils.mapperParser("hiveColumnMetaStore", params).toString();
    }

    @Override
    public ColumnMetaData columnMetaDataBuilder(ResultSet resultSet) throws SQLException {
        ColumnMetaData columnMetaData = new ColumnMetaData();
        columnMetaData.setColumnName(resultSet.getString("column_name"));
        Tuple3<String, Integer, Integer> t = RegexMatchUtils.columnNameAnalysis(resultSet.getString("original_column_type"));
        columnMetaData.setOriginalColumnType(t.f0);
        columnMetaData.setColumnPrecision(t.f1);
        columnMetaData.setColumnScale(t.f2);
        columnMetaData.setColumnComment(resultSet.getString("column_comment"));
        return columnMetaData;
    }
}
